package net.bwie.order.dwd.job;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.order.dwd.function.LoadProvinceDimMapFunction;
import net.bwie.order.dwd.function.RequestDicDimAsyncFunction;
import net.bwie.realtime.guanjuntao.util.DateTimeUtil;
import net.bwie.realtime.guanjuntao.util.DorisUtil;
import net.bwie.realtime.guanjuntao.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/*
交易-下单明细-事务事实表：
从Kafka读取订单相关表数据和HBase读取相关维表数据，进行关联JOIN，构建订单明细宽表数据，写入Doris表；
 */
public class JtpOrderDetailDorisDwdJob {
    public static void main(String[] args) throws Exception {

        // TODO: 2025/6/5 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // TODO: 2025/6/5 数据源-source
        DataStream<String> dataStream = KafkaUtil.consumerKafka(env, "topic-db");
        //dataStream.print();
        // TODO: 2025/6/5 数据转换-transformation
        DataStream<String> dwdStream = handle(dataStream);
        dwdStream.print();
        // TODO: 2025/6/5 数据sink-sink
        DorisUtil.saveToDoris(dwdStream, "jtp_mall_doris", "dwd_order_detail");
        // TODO: 2025/6/5 触发执行
        env.execute("JtpOrderDetailDorisDwdJob");


    }

    /*
    订单数据关联,添加维度信息,形成订单明细宽表,最后写入Doris表,基于物化视图实时数据聚合
     数据流join采用interval join
     省份维度补全采用预加载的方式
     字典维度补全采用异步IO的方式
     */
    private static DataStream<String> handle(DataStream<String> dataStream) {
        // TODO: 2025/6/5 订单明细流关订单信息流 
        DataStream<String> joinStream = joinProcss(dataStream);
        // TODO: 2025/6/5 省份维度补全  预加载
        DataStream<String> provinceStream = joinStream.map(new LoadProvinceDimMapFunction());
        // TODO: 2025/6/5 字典维度补全  异步IO
 // 创建一个有序等待的数据流，用于异步获取字典维度信息
DataStream<String> dwdStream = AsyncDataStream.orderedWait(
        provinceStream, // 输入的数据流，包含省份信息
        new RequestDicDimAsyncFunction(), // 自定义的异步函数，用于请求字典维度信息
        10000, // 最大等待时间，单位为毫秒
        TimeUnit.MILLISECONDS, // 时间单位
        100 // 最大缓冲区大小，用于存储等待中的结果
);

        return dwdStream;
    }

    private static DataStream<String> joinProcss(DataStream<String> dataStream) {
        /*
        获取数据流
         */

        // TODO: 2025/6/5 数据明细流
        SingleOutputStreamOperator<String> detailStreame = dataStream.filter(new RichFilterFunction<String>() {
            @Override
            public boolean filter(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                return "order_detail".equals(jsonObject.getString("table_name"))
                        && "insert".equals(jsonObject.getString("operate_type"));
            }
        });
        // TODO: 2025/6/5 订单信息流
        SingleOutputStreamOperator<String> infoStream = dataStream.filter(new RichFilterFunction<String>() {
            @Override
            public boolean filter(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                return "order_info".equals(jsonObject.getString("table_name"))
                        && "insert".equals(jsonObject.getString("operate_type"));
            }
        });

        /*
        指定事件时间和watermark水位线
         */
        // TODO: 2025/6/5 订单明细流
        SingleOutputStreamOperator<String> timeDetailStreame = detailStreame.assignTimestampsAndWatermarks(
                WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner(new SerializableTimestampAssigner<String>() {
                            @Override
                            public long extractTimestamp(String element, long recordTimestamp) {
                                String eventTime = JSON.parseObject(element)
                                        .getJSONObject("operate_data").getString("create_time");

                                return DateTimeUtil.convertStringToLong(eventTime, "yyyy-MM-dd HH:mm:ss");
                            }
                        })
        );
        // TODO: 2025/6/5 订单信息流
        SingleOutputStreamOperator<String> timeInfoStream = infoStream.assignTimestampsAndWatermarks(
                WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner(new SerializableTimestampAssigner<String>() {


                            @Override
                            public long extractTimestamp(String element, long recordTimestamp) {
                                String eventTime = JSON.parseObject(element)
                                        .getJSONObject("operate_data").getString("create_time");
                                return DateTimeUtil.convertStringToLong(eventTime, "yyyy-MM-dd HH:mm:ss");
                            }
                        })
        );

        /*
          数据关联join,由于info关联detail,所以使用interval join
          interval join 要求2个数据为分组流(keyed staeam) 按照关联字段分组keyBy
         */

        SingleOutputStreamOperator<String> joinStream = timeInfoStream
                .keyBy(element -> JSON.parseObject(element).getJSONObject("operate_data").getString("id"))
                .intervalJoin(
                        timeDetailStreame.keyBy(element -> JSON.parseObject(element).getJSONObject("operate_data").getString("order_id")))
                .between(Time.seconds(-10), Time.seconds(10))
                .process(new ProcessJoinFunction<String, String, String>() {
                    @Override
                    public void processElement(String info, String detail,
                                               Context ctx, Collector<String> out) throws Exception {
                        // TODO: 2025/6/5 获取user_id和province_id
                        JSONObject infoJsonObject = JSON.parseObject(info).getJSONObject("operate_data");
                        Integer user_id = infoJsonObject.getInteger("user_id");
                        Integer province_id = infoJsonObject.getInteger("province_id");
                        // TODO: 2025/6/5 将user_id和province_id添加到detail中
                        JSONObject detailJsonObject = JSON.parseObject(detail).getJSONObject("operate_data");
                        detailJsonObject.put("user_id", user_id);
                        detailJsonObject.put("province_id", province_id);
                        // TODO: 2025/6/5 订单日期
                        String createTime = detailJsonObject.getString("create_time");
                        String substring = createTime.substring(0, 10);
                        detailJsonObject.put("cur_date", substring);
                        // TODO: 2025/6/5 输出
                        String output = detailJsonObject.toJSONString();
                        out.collect(output);
                    }
                });
        return joinStream;





    }


















}
